package redis

import (
	"errors"
	"fmt"
	"github.com/gomodule/redigo/redis"
	"log"
	"strconv"
	"strings"
)

const (
	kClusterSlots = 16384
)

//redis-cluster
type (
	Cluster struct {
		debug            bool
		attempts         int                  //重定向次数
		clusterNodes     []*Node              //集群节点列表
		slots            [kClusterSlots]*Node //槽节点对应信息
		saveSlotsRunning bool                 //本地化槽和节点信息正在进行
	}
)

func NewCluster(address ...string) *Cluster {
	this := &Cluster{
		attempts: 5,
	}

	for _, v := range address {
		this.clusterNodes = append(this.clusterNodes, &Node{Address: v})
	}

	this.initSlots()

	return this

}

func (this *Cluster) Debug() *Cluster {
	this.debug = true
	return this
}

func (this *Cluster) SetAttempts(num int) *Cluster {
	this.attempts = num
	return this
}

//配置详细的节点信息
func (this *Cluster) SetNodes(nodes ...*Node) *Cluster {
	readOrCreateNode := func(node *Node) {
		for k := range this.clusterNodes {
			if this.clusterNodes[k].Address == node.Address {
				//更新
				this.clusterNodes[k] = node
				return
			}
		}

		//添加
		this.clusterNodes = append(this.clusterNodes, node)
		return
	}

	for k := range nodes {
		readOrCreateNode(nodes[k])
	}

	return this

}

func (this *Cluster) Do(commandName string, args ...interface{}) (reply interface{}, err error) {
	var (
		key        string
		node       *Node
		lastAttemp int
	)
	if len(args) < 1 {
		return nil, fmt.Errorf("no key found in args")
	}

	if key, err = this.toString(args[0]); err != nil {
		return nil, fmt.Errorf("invalid key %v", args[0])
	}

	if node, err = this.getNodeByKey(key); err != nil {
		return
	}

	if this.debug {
		log.Printf("通过key从本地缓存中获取的节点为：%s,slot:%d\n", node.Address, this.hashSlot(key))
		log.Printf("获取solts[%d]:%p", this.hashSlot(key), this.slots[this.hashSlot(key)])
	}

	//node=this.getNodeByAddress("172.50.0.1:6392")
	reply, err, lastAttemp = this.runWithRetries(node, commandName, args, this.attempts)
	if err == nil && lastAttemp < this.attempts {
		//发生了重定向，并成功执行命令
		this.initSlots()
		log.Printf("发生重定向,solts[%d]:%p", this.hashSlot(key), this.slots[this.hashSlot(key)])

	}

	return
}

//为每个节点创建redisPool
func (this *Cluster) initRedisPools() *Cluster {
	for k := range this.clusterNodes {
		this.clusterNodes[k].NewRedisPool()
	}

	return this
}

//从集群中选一个可运行的节点，使用cluster slots初始化槽和节点映射
func (this *Cluster) initSlots() {
	var (
		find bool
		err  error
	)

	if this.saveSlotsRunning {
		return
	}

	this.saveSlotsRunning = true

	defer func() {
		if err != nil {
			log.Println("初始化槽和节点缓存结束,错误：", err.Error())
		}

		if this.debug {
			log.Println("初始化槽和节点缓存结束")
		}
	}()

	if this.debug {
		log.Println("初始化槽和节点缓存开始")
	}

	for k := range this.clusterNodes {

		if err := this.saveSlots(this.clusterNodes[k].Conn()); err == nil {
			if this.debug {
				log.Printf("已从节点[%s],缓存slots\n", this.clusterNodes[k].Address)
			}
			find = true
			break
		}
	}

	if !find {
		err = errors.New("No cluster nodes available")
	}

	this.saveSlotsRunning = false
}

//从节点中获取slots和node的信息，并缓存到本地
func (this *Cluster) saveSlots(c redis.Conn) (err error) {
	var clusterSlotsRes []interface{}
	defer c.Close()

	if c.Err() != nil {
		return c.Err()
	}

	if clusterSlotsRes, err = redis.Values(c.Do("Cluster", "Slots")); err != nil {
		return
	}

	for _, v := range clusterSlotsRes {
		ss := ToSlice(v)
		if len(ss) >= 3 {
			startSlot := ss[0].(int64)
			endSlot := ss[1].(int64)

			for k, v := range ss[2:] {
				nodeInfo := ToSlice(v)
				if len(nodeInfo) != 3 {
					continue
				}
				//获取节点数据
				node := this.readOrCreateNode(fmt.Sprintf("%s:%v", nodeInfo[0], nodeInfo[1]))

				if k == 0 {
					for startSlot <= endSlot {
						this.slots[startSlot] = node
						startSlot++
					}
				}
			}
		}
	}

	return
}

//读取或者创建节点
func (this *Cluster) readOrCreateNode(address string) *Node {
	for k := range this.clusterNodes {
		if this.clusterNodes[k].Address == address {
			return this.clusterNodes[k]
		}
	}

	node := &Node{
		Address: address,
	}

	this.clusterNodes = append(this.clusterNodes, node)

	return node
}

func (this *Cluster) toString(arg interface{}) (string, error) {
	switch arg := arg.(type) {
	case int:
		return strconv.Itoa(arg), nil
	case int64:
		return strconv.Itoa(int(arg)), nil
	case float64:
		return strconv.FormatFloat(arg, 'g', -1, 64), nil
	case string:
		return arg, nil
	case []byte:
		return string(arg), nil
	default:
		return "", fmt.Errorf("unknown type %T", arg)
	}
}

func (this *Cluster) hashSlot(key string) uint16 {
	var s, e int
	for s = 0; s < len(key); s++ {
		if key[s] == '{' {
			break
		}
	}

	if s == len(key) {
		return crc16(key) & (kClusterSlots - 1)
	}

	for e = s + 1; e < len(key); e++ {
		if key[e] == '}' {
			break
		}
	}

	if e == len(key) || e == s+1 {
		return crc16(key) & (kClusterSlots - 1)
	}

	return crc16(key[s+1:e]) & (kClusterSlots - 1)
}

//递归执行，发生MOVED,ASKing,数据迁移过程中遇到问题
func (this Cluster) runWithRetries(node *Node, commandName string, args []interface{}, attempts int) (reply interface{}, err error, lastAttempts int) {
	lastAttempts = attempts
	if attempts <= 0 {
		err = errors.New("Too many Cluster redirections?")
		return
	}

	if node == nil {
		err = errors.New("No cluster nodes available")
		return
	}

	c := node.Conn()
	defer c.Close()

	if c.Err() == nil { //连接正常
		//执行命令
		if reply, err = c.Do(commandName, args...); err == nil {
			//检查replay
			if node = this.getRedirectNode(reply, node); node == nil {
				//命令执行成功,正常退出
				if this.debug {
					log.Printf("执行成功,重试次数:%d,命令:%s%v\n", this.attempts-lastAttempts, commandName, args)
				}
				return
			}
		} else {
			//检查错误
			//执行命令出现问题,检查是否有重定向
			if node = this.getRedirectNode(err.Error(), node); node == nil {
				return
			}
		}

	} else {
		//连接异常,随机获取可用node
		node = this.getAvailableNode(node.Address)

	}

	return this.runWithRetries(node, commandName, args, attempts-1)
}

//通过key 获取一个没有连接问题的节点
func (this *Cluster) getNodeByKey(key string) (node *Node, err error) {

	//查询不到槽对应的节点，重新初始化
	if node = this.slots[this.hashSlot(key)]; node == nil {
		for k := range this.clusterNodes {

			if this.clusterNodes[k] != nil {
				c := this.clusterNodes[k].Conn()
				defer c.Close()
				if c.Err() != nil {
					//连接有问题
					continue
				}

				node = this.clusterNodes[k]
				break
			}

		}
	}

	if node == nil {
		return nil, errors.New("No cluster nodes available")
	}

	return
}

//顺序获取一个可用节点
func (this *Cluster) getAvailableNode(excludeAddress string) *Node {
	for k := range this.clusterNodes {

		//排除节点
		if this.clusterNodes[k].Address == excludeAddress {
			continue
		}

		if this.clusterNodes[k] != nil {
			c := this.clusterNodes[k].Conn()
			defer c.Close()
			if c.Err() != nil {
				//连接有问题
				continue
			}

			return this.clusterNodes[k]
		}

	}
	return nil
}

//从响应中解析出地址
func (this *Cluster) parseAddress(replyMsg string) (address string) {
	fields := strings.Split(replyMsg, " ")
	if len(fields) != 3 {
		return
	}

	return fields[2]
}

//获取目标重定向节点
func (this *Cluster) getRedirectNode(reply interface{}, curNode *Node) *Node {
	var (
		replyMsg string
		ok       bool
	)
	if replyMsg, ok = reply.(string); !ok {
		return nil
	}
	address := this.parseAddress(replyMsg)
	switch {
	case strings.HasPrefix(replyMsg, "ASK"):
		return this.getNodeByAddress(address)
	case strings.HasPrefix(replyMsg, "MOVED"):
		return this.getNodeByAddress(address)
	case strings.HasPrefix(replyMsg, "ECONNTIMEOUT"):
		return this.getAvailableNode(curNode.Address)

	}
	return nil
}

func (this *Cluster) getNodeByAddress(address string) *Node {
	for k := range this.clusterNodes {
		if this.clusterNodes[k].Address == address {
			return this.clusterNodes[k]
		}
	}

	return nil
}
